package com.qyer.hive.udf;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.List;
import java.util.Map;
import java.util.PropertyResourceBundle;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.codehaus.jackson.map.ObjectMapper;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

public class IpforCity  extends UDF {
	private static List<GeoBlockModel> geoList = Lists.newArrayListWithExpectedSize(2005200);
	private static Map<Integer, IPFinalModel> ipFinalMap = Maps.newHashMapWithExpectedSize(567382);
	private final ObjectMapper objectMapper = new ObjectMapper();
	static{
		String geoFilePath = null;
		String ipFinalFilePath = null;
		ClassLoader cl = IpforCity.class.getClassLoader();
		InputStream inputStream = null;
		try{
			inputStream = cl.getResourceAsStream( "config.properties" );
			if( inputStream != null ){
				PropertyResourceBundle propertyResourceBundle = new PropertyResourceBundle(inputStream );
				geoFilePath = propertyResourceBundle.getString("geoFile");
				ipFinalFilePath = propertyResourceBundle.getString("ipFinalFile");
			}
		}catch( Exception e ){
			e.printStackTrace();
		}finally{
			if( inputStream != null ){
				try{
					inputStream.close();
				}
				catch( Exception e2 ){
				}
			}
		}
		loadGeoData(geoFilePath);
		loadCityData(ipFinalFilePath);
	}
	
	/**
	 * 加载地理信息
	 */
	public static void loadGeoData(String geoFilePath) {
		BufferedReader br = null;
		try{
			InputStream inputStream = new FileInputStream(geoFilePath);
			InputStreamReader fr = new InputStreamReader(inputStream, "UTF-8");
			br = new BufferedReader(fr);
			String line = br.readLine();
			while (line != null) {
				geoList.add(new GeoBlockModel(line));
				line = br.readLine();
			}
			br.close();
		}catch(IOException e){
			e.printStackTrace();
		}
	}
	
	/**
	 * 加载城市信息
	 */
	public static void loadCityData(String ipFinalFilePath) {
		BufferedReader br = null;
		try{
			InputStream inputStream = new FileInputStream(ipFinalFilePath);
			InputStreamReader fr = new InputStreamReader(inputStream, "UTF-8");
			br = new BufferedReader(fr);
			String line = br.readLine();
			while (line != null) {
				String[] arr = line.split(",", -1);
				ipFinalMap.put(Integer.parseInt(arr[0]),new IPFinalModel(Integer.parseInt(arr[10]), arr[3], Integer.parseInt(arr[9]), arr[1]));
				line = br.readLine();
			}
			br.close();
		}catch(IOException e){
			e.printStackTrace();
		}
	}
	
	private static final String rexp = "^([1-9]|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])(\\.(\\d|[1-9]\\d|1\\d{2}|2[0-4]\\d|25[0-5])){3}$";
	private static Pattern pat = Pattern.compile(rexp); 
	
	public static  boolean isIP(String addr)
    {
		Matcher mat = pat.matcher(addr); 
		boolean ipAddress = mat.matches();
		return ipAddress;
    }
	
	/**
	 * IP地址转整数
	 * @param ipAddress
	 * @return
	 */
	public static long computeIP(String ipAddress){
	 long integer_ip = -1;
	 if(isIP(ipAddress)){
		 String[] arr = ipAddress.split("\\.");
		 integer_ip = 16777216L * Integer.parseInt(arr[0]) + ( 65536 * Integer.parseInt(arr[1]) ) + ( 256 * Integer.parseInt(arr[2]) ) + Integer.parseInt(arr[3]);
	 }
	 return integer_ip;
	}	    
	   
	/**
	 * 根据整型IP查索引
	 * @param ipID
	 * @return
	 */
	private static int binarySearch(long ipID){
		int low =0,mid=0;
		int high = geoList.size() - 1;
		GeoBlockModel midVo = null;
		while(low <= high){
			mid = (low + high)/2;
			midVo = geoList.get(mid); 
			if(midVo.getLat() < ipID){
				low = mid + 1;
			}else if(midVo.getLon() > ipID){
				high = mid - 1;
			}else{
				return midVo.getIpID();
			}
		}
		return 0;
	}
 		
	public String evaluate(final String s) {  
        if (s == null) { return "{}"; }  
        long ip = computeIP(s.trim());
        if(ip < 0){
        	return "{}";
        }
        int index = binarySearch(ip); 
        IPFinalModel cityInfo = ipFinalMap.get(index);
        if(cityInfo == null){
        	return "{}";
        }
        try {
			return objectMapper.writeValueAsString(cityInfo);
		} catch (IOException e) {
			e.printStackTrace();
			return "{}";
		} 
    }  
	
	public static void main(String[] args) {
//		IpforCity ipforCity = new IpforCity();
//		String result = ipforCity.evaluate("117.25.124.144");
//		String twoAddress = "127.30.45.257";
//		System.out.println(isIP(twoAddress));
	}

}
